package com.wenwen.social.domain.service.Impl;

import com.wenwen.social.domain.model.FollowLine;
import com.wenwen.social.domain.model.PersonLine;
import com.wenwen.social.domain.service.FollowLineService;
import com.wenwen.social.domain.service.PersonLineService;
import com.wenwen.social.util.PageUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

import static com.wenwen.social.constant.FollowLineConstant.FOLLOW_LINE_REDIS_PRE;
import static java.util.stream.Collectors.toList;

@Service
public class FollowLineServiceImpl implements FollowLineService {
	@Autowired
	private RedisTemplate redisTemplate;

	@Override
	public void add(Long uid,FollowLine follow) {
		ZSetOperations zSetOperations = redisTemplate.opsForZSet();
		zSetOperations.add(getKey(uid),follow.getMessageId(), follow.getTime());
		return;
	}

	@Override
	public void add(Set<Long> uids, FollowLine followLine) {
		redisTemplate.executePipelined(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
				for (Long uid : uids) {
					redisConnection.zAdd(getKey(uid).getBytes(),followLine.getTime(),(followLine.getMessageId()+"").getBytes());
				}
				return null;
			}
		});
	}

	@Override
	public List<FollowLine> query(Long uid,Integer page, Integer num) {
		ZSetOperations zSetOperations = redisTemplate.opsForZSet();
		PageUtil pageUtil = new PageUtil(page, num);
		Set<DefaultTypedTuple> range = zSetOperations.reverseRangeWithScores(getKey(uid), pageUtil.getStartNum(), pageUtil.getEndNum());
		if(range == null || range.isEmpty()){
			return new ArrayList<>();
		}

		List<FollowLine> result = range.stream()
				.map(t -> new FollowLine((Long) t.getValue(), t.getScore().longValue()))
				.collect(toList());
		return result;
	}

	@Override
	public Integer del(Long uid, Long msgId) {
		ZSetOperations zSetOperations = redisTemplate.opsForZSet();
		return zSetOperations.remove(getKey(uid), msgId).intValue();
	}

	@Override
	public Integer del(Long uid,List<Long> collect) {
		ZSetOperations zSetOperations = redisTemplate.opsForZSet();
		Long remove = zSetOperations.remove(getKey(uid), collect.toArray());
		return remove.intValue();
	}

	@Override
	public Integer del(Set<Long> uids, Long msgId) {
		List list = redisTemplate.executePipelined(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
				for (Long uid : uids) {
					redisConnection.zRem(getKey(uid).getBytes(), (msgId+"").getBytes());
				}
				return null;
			}
		});
		return list.size();
	}

	private String getKey(Long uid) {
		return FOLLOW_LINE_REDIS_PRE+uid;
	}

	@Override
	public Integer limit(Long uid, Integer index, Integer size) {
		ZSetOperations zSetOperations = redisTemplate.opsForZSet();
		return zSetOperations.removeRange(getKey(uid),index,size-1).intValue();
	}

	@Override
	public Long count(Long uid) {
		String key = getKey(uid);
		ZSetOperations zSetOperations = redisTemplate.opsForZSet();
		return zSetOperations.zCard(key);
	}
}
